// Copyright (C) 2012 Yury Bychenkov (http://bychenkov.com).
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
//  - Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
//  - Redistributions in binary form must reproduce the above
// copyright notice, this list of conditions and the following disclaimer
// in the documentation and/or other materials provided with the
// distribution.
//  - Neither the name of Yury Bychenkov nor the names of its
// contributors may be used to endorse or promote products derived from
// this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

#include "database.h"
#include "filesystem.h"
#include "numberutil.h"

Logger Database::logger("Database");

RecordBuilder::RecordBuilder(const RecordBuilder& that):
	schema(that.schema),
	bundle(that.bundle),
	currentColumnIndex(that.currentColumnIndex),
	bad(that.bad)
{
}

RecordBuilder::RecordBuilder(const Schema& schema_):
	schema(schema_),
	currentColumnIndex(0),
	bad(false)
{
}

bool RecordBuilder::isBad() const {
	return bad;
}

bool RecordBuilder::isComplete() const {
	return currentColumnIndex == schema.getColumnCount();
}

RecordBuilder& RecordBuilder::add(const char* value) {
	if (value == 0) {
		return addNull();
	}
	if (!bad && currentColumnIndex < schema.getColumnCount()) {
		uint type = schema.getColumn(currentColumnIndex).type;

		if (type == Schema::STRING) {
			addString(value, (uint) strlen(value));
		} else if (type == Schema::DECIMAL) {
			if (value[0] == '\0') {
				return addNull();
			}

			const char* end = 0;
			uint8 buffer[0xFE];

			int size = NumberUtil::parseDecimal(value, buffer, sizeof(buffer), &end);
			if (size > 0 && *end == 0) {
				addDecimal(buffer, (uint) size);
			} else {
				bad = true;
			}
		} else if (type == Schema::INTEGER) {
			if (value[0] == '\0') {
				return addNull();
			}

			bad = true;
			int64 v;
			const char* end = 0;

			if (NumberUtil::parseInteger(value, v, &end)) {
				if (*end == '.') {
					do {
						end++;
					} while (*end == '0');
				}
			
				if (*end == 0) {
					bad = false;
					addInteger(v);
				}
			}
		} else {
			bad = true;
		}
	}

	return *this;
}

RecordBuilder& RecordBuilder::addNull() {
	if (!bad && currentColumnIndex < schema.getColumnCount()) {
		bad = !bundle.addNull();
		currentColumnIndex++;
	}
	return *this;
}

RecordBuilder& RecordBuilder::addString(const void* value, uint size) {
	if (!bad && currentColumnIndex < schema.getColumnCount()) {
		uint columnType = schema.getColumn(currentColumnIndex).type;
		if (columnType == Schema::STRING) {
			bad = !bundle.addString(value, size);
			currentColumnIndex++;
		} else {
			bad = true;
		}
	}
	return *this;
}

RecordBuilder& RecordBuilder::addDecimal(const void* value, uint size) {
	if (!bad && currentColumnIndex < schema.getColumnCount()) {
		uint columnType = schema.getColumn(currentColumnIndex).type;
		if (columnType == Schema::DECIMAL) {
			bad = !bundle.addDecimal(value, size);
			currentColumnIndex++;
		} else {
			bad = true;
		}
	}
	return *this;
}

RecordBuilder& RecordBuilder::addInteger(int64 value) {
	if (!bad && currentColumnIndex < schema.getColumnCount()) {
		uint columnType = schema.getColumn(currentColumnIndex).type;
		if (columnType == Schema::INTEGER) {
			bad = !bundle.addInteger(value);
			currentColumnIndex++;
	   	} else {
			bad = true;
		}
	}
	return *this;
}

void RecordBuilder::reset() {
	bundle.clear();
	currentColumnIndex = 0;	
	bad = false;
}

Database::Database(const DatabaseManager& dbManager_, const string& path_, const Schema& schema_, const vector<Database::Chunk>& chunks_):
	dbManager(dbManager_),
	path(path_),
	schema(schema_),
	chunks(chunks_)
{
}

Database::~Database() {

	Chunk chunk;
	for (uint i = 0; chunks.get(i, chunk); i++) {
		if (chunk.nunChunk != 0) {
			delete chunk.nunChunk;
		}
		if (chunk.logChunk != 0) {
			delete chunk.logChunk;
		}
	}
	
}

RecordBuilder Database::newRecordBuilder() {
	return RecordBuilder(schema);
}

const Schema& Database::getSchema() const {
	return schema;
}

string Database::getNunFileName(uint index) {
	return string("nun") + (index == 0? "*": NumberUtil::formatHex(index, 8));
}

string Database::getLogFileName(uint index) {
	return string("log") + (index == 0? "*": NumberUtil::formatHex(index, 8));
}

Database* Database::open(const DatabaseManager& dbManager, const char* path_) {
	string path(path_);

	map<uint, string> files;

	vector<string> pathFiles = FileSystem::list(path);
	for (vector<string>::iterator it = pathFiles.begin(); it != pathFiles.end(); it++) {
		uint index;
		if (it->size() > 3 && 0 == strncmp("log", it->c_str(), 3) && (stringstream(it->c_str() + strlen("log")) >> hex >> index).eof()) {
			if (*it == getLogFileName(index)) {
				files[index] = *it;
			}
		}
	}

	for (vector<string>::iterator it = pathFiles.begin(); it != pathFiles.end(); it++) {
		uint index;
		if (it->size() > 3 && 0 == strncmp("nun", it->c_str(), 3) && (stringstream(it->c_str() + strlen("nun")) >> hex >> index).eof()) {
			if (*it == getNunFileName(index)) {
				files[index] = *it;
			}
		}
	}

	path += "/";

	string schemaFile = path + "schema";
	Schema schema = Schema::load(schemaFile.c_str());
	if (schema.getColumnCount() == 0) {
		logger.error("Schema unavailable");
		return 0;
	}

	vector<Chunk> chunks;

	for (map<uint, string>::iterator it = files.begin(); it != files.end(); it++) {
		Chunk chunk;
		chunk.logChunk = 0;
		chunk.nunChunk = 0;
		chunk.index = it->first;

		string chunkFile = path + it->second;
		if (it->second == getNunFileName(it->first)) {
			chunk.nunChunk = NunChunk::open(chunkFile.c_str());
			if (chunk.nunChunk != 0) {
			 	chunks.push_back(chunk);
				logger.info("NunChunk mounted: ", it->second);
				continue;
			}
		} else if (it->second == getLogFileName(it->first)) {
			bool readOnly = it != --files.end();
			chunk.logChunk = LogChunk::open(chunkFile.c_str(), readOnly);
			if (chunk.logChunk != 0) {
			 	chunks.push_back(chunk);
				logger.info("LogChunk mounted: ", it->second, readOnly? " (ro)": " (rw)");
				continue;
			}
		} 

		logger.error("Broken chunk: ", it->second);

		for (uint i = 0; i < chunks.size(); i++) {
			if (chunks[i].logChunk != 0) {
				delete chunks[i].logChunk;
			}
			if (chunks[i].nunChunk != 0) {
				delete chunks[i].nunChunk;
			}
		}
	
		return 0;
	}

	return new Database(dbManager, path, schema, chunks);
}


bool Database::addChunk() {
	Chunk chunk;

	chunk.index = chunks.getCount() == 0 ? 1 : (chunks[chunks.getCount() - 1].index + 1);

	string chunkFile = path + getLogFileName(chunk.index);

	chunk.logChunk = LogChunk::create(chunkFile.c_str(), schema.getColumnCount(), 65536, 4096); // TODO config
	if (chunk.logChunk == 0) {
		logger.error("Can't map a new chunk: ", getLogFileName(chunk.index));
		return false;
	}

	chunk.nunChunk = 0;
	logger.info("LogChunk mapped (rw): ", getLogFileName(chunk.index));

	chunks.add(chunk);
	
	return true;
}

bool Database::add(const RecordBuilder& recordBuilder) {

	if (recordBuilder.isBad() || !recordBuilder.isComplete()) {
		return false;
	}

	for (int attempt = 0; attempt < 2; attempt++) {
		Chunk chunk;
		chunk.logChunk = 0;

		if (!chunks.getLast(chunk) || chunk.logChunk == 0) {
			attempt++;
		}
		if (attempt == 1) {
			if (!addChunk()) {
				return false;
			}
			chunks.getLast(chunk);
		}

		LogChunk* logChunk = chunk.logChunk;

		logChunk->begin();

		const ResourceBundle& bundle = recordBuilder.bundle;
		bool success = true;
		for (uint columnIndex = 0, columnCount = schema.getColumnCount(); columnIndex < columnCount; columnIndex++) {
			uint size = 0;
			const uint8* element = bundle.get(columnIndex, size);

			if (size == 0) {
				return false;
			}

			if (!logChunk->add(element, size)) {
				success = false;
				break;
			}
		}

		if (success) {
			if (logChunk->commit()) {
				return true;
			}
		}
    }

    return false;
}


void Database::manage() {
	uint count = chunks.getCount();
	
	Chunk chunk;
	
	for (uint i = 0; i + 1 < count && chunks.get(i, chunk); i++) {
		if (chunk.nunChunk == 0) {
			string nunChunkFile = path + getNunFileName(chunk.index);
			string nunChunkFileTmp = nunChunkFile + ".tmp";
			logger.info("LogChunk ", getLogFileName(chunk.index), " is being preprocessed ...");
			
			clock_t startTime = clock();

			if (NunChunk::build(schema, nunChunkFileTmp.c_str(), *chunk.logChunk)) {
				if (FileSystem::move(nunChunkFileTmp.c_str(), nunChunkFile.c_str())) {
					chunk.nunChunk = NunChunk::open(nunChunkFile.c_str());
					if (chunk.nunChunk != 0) {
						chunks.set(i, chunk);
						uint dt = uint((clock() - startTime) * 1000 / CLOCKS_PER_SEC);
						stringstream ss;
						ss << dt / 1000 << "." << dt % 1000 << " secs";
						logger.info(getNunFileName(chunk.index), " successfully created in ", ss.str());
					}
				} else {
					logger.error("Can't rename ", nunChunkFileTmp);
				}
			} else {
				logger.error(getLogFileName(chunk.index), " preprocessing failed");
			}

			return;
		}
	}
}
